package com.freestyle.jmssample.rabbitmq.service;

import com.freestyle.jmssample.api.ReceiveSMSService;
import com.freestyle.jmssample.api.SMSConstants;
import com.freestyle.jmssample.api.pojo.JSONData;
import com.freestyle.jmssample.api.pojo.SMSBean;
import com.freestyle.jmssample.rabbitmq.SMSUtil;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.Date;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

/**
 * SMS接收实例
 * Created by rocklee on 2022/1/6 10:30
 */
@Service
@Slf4j
public class ReceiveSMSServiceImp implements ReceiveSMSService {
  private BiConsumer<String,SMSBean> consumer;

  //onReceived、onReceived2会同时收到消息
  @RabbitListener(bindings = @QueueBinding(value=@Queue(SMSConstants.SMS_QUEUE_DIRECT),exchange = @Exchange(value = SMSConstants.SMS_EXCHANGE_DIRECT,
          type = ExchangeTypes.DIRECT),key = SMSConstants.SMS_KEY_DIRECT))
  @RabbitHandler
  protected void onReceived(Message message,SMSBean sms, Channel channel) throws IOException {
     log.info("============onReceived接收=========");
     if (consumer!=null) {
       consumer.accept(SMSConstants.SMS_KEY_DIRECT+"-"+SMSConstants.SMS_QUEUE_DIRECT, sms);
     }
     //channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
  }
  @RabbitListener(bindings = @QueueBinding(value=@Queue(SMSConstants.SMS_QUEUE_DIRECT2),exchange = @Exchange(value = SMSConstants.SMS_EXCHANGE_DIRECT,
          type = ExchangeTypes.DIRECT),key = SMSConstants.SMS_KEY_DIRECT))
  @RabbitHandler
  protected void onReceived2(Message message,SMSBean sms, Channel channel) throws IOException {
    log.info("============onReceived2接收=========");
    if (consumer!=null) {
      consumer.accept(SMSConstants.SMS_KEY_DIRECT+"-"+SMSConstants.SMS_QUEUE_DIRECT2, sms);
    }
   // channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
  }

  @RabbitListener(ackMode = "MANUAL",bindings = @QueueBinding(value=@Queue(SMSConstants.SMS_QUEUE_DIRECT3),exchange = @Exchange(value = SMSConstants.SMS_EXCHANGE_DIRECT,
          type = ExchangeTypes.DIRECT),key = SMSConstants.SMS_KEY_DIRECT3))
  @RabbitHandler
  protected void onReceived3(Message message,SMSBean sms, Channel channel) throws IOException, InterruptedException {
    log.info("===========onReceived3收到消息(手动ACK)========");
    if (consumer!=null) {
      consumer.accept(SMSConstants.SMS_KEY_DIRECT3+"-"+SMSConstants.SMS_QUEUE_DIRECT3, sms);
    }
    //注意，cusumer的这些应答只是跟broker(MQ中间件）之间进行，并没影响到producer。
    if (sms.getSubject().equalsIgnoreCase("1")){ //
      channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);//确认
    }
    else if (sms.getSubject().equalsIgnoreCase("2")){
      //basicNack(boolean multiple, boolean requeue)
      channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);//不确认,丢弃
    }
    else {
      //channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);//不确认，丢弃
      if (message.getMessageProperties().getRedelivered()){ //已经重投的就放弃了
        log.info("此为重投消息，直接删除："+sms);
        channel.basicReject(message.getMessageProperties().getDeliveryTag(),false);
      }
      else {
        log.info("不确认，将消息重新投入队列:"+sms);
        channel.basicReject(message.getMessageProperties().getDeliveryTag(), true); //不确认，并且重投
      }
    }
  }
  @RabbitListener(bindings = @QueueBinding(value=@Queue(SMSConstants.SMS_QUEUE_DIRECT4),exchange = @Exchange(value = SMSConstants.SMS_EXCHANGE_DIRECT,
          type = ExchangeTypes.DIRECT),key = SMSConstants.SMS_KEY_DIRECT))
  @RabbitHandler
  //应答
  protected JSONData<Object> onReceived4(Message message, SMSBean sms, Channel channel) throws IOException {
     log.info("=========onReceived4已接收，返回数据===========");
     SMSUtil.sleep(5000);
     JSONData<Object> ret=JSONData.JSONDataFromResult(String.format("%s,消费端已接收%s",new Date().toString(),sms));
     return  ret;
  }


  //订阅接收 通过topic交换机发出的匹配SMS_KEY_TOPIC键值的都能接收
  @RabbitListener(bindings = @QueueBinding(value=@Queue(SMSConstants.SMS_QUEUE_TOPIC),exchange = @Exchange(value = SMSConstants.SMS_EXCHANGE_TOPIC,
          type = ExchangeTypes.TOPIC),key = SMSConstants.SMS_KEY_TOPIC))
  @RabbitHandler
  protected void onBroadcastReceive(Message message, SMSBean sms, Channel channel) throws IOException {
    log.info("=========onBroadcastReceive已接收===========");
    if (consumer!=null) {
      consumer.accept(SMSConstants.SMS_KEY_TOPIC+"-"+SMSConstants.SMS_QUEUE_TOPIC, sms);
    }
  }
  //订阅接收
  @RabbitListener(bindings = @QueueBinding(value=@Queue(SMSConstants.SMS_QUEUE_FANOUT_1),exchange = @Exchange(value = SMSConstants.SMS_EXCHANGE_FANOUT,
          type = ExchangeTypes.FANOUT)))
  @RabbitHandler
  protected void onFanOutReceive1(Message message, SMSBean sms, Channel channel) throws IOException {
    log.info("=========onFanOutReceive1已接收===========");
    if (consumer!=null) {
      consumer.accept("FanOut-"+SMSConstants.SMS_QUEUE_FANOUT_1, sms);
    }
  }
  //订阅接收
  @RabbitListener(bindings = @QueueBinding(value=@Queue(SMSConstants.SMS_QUEUE_FANOUT_2),exchange = @Exchange(value = SMSConstants.SMS_EXCHANGE_FANOUT,
          type = ExchangeTypes.FANOUT)))
  @RabbitHandler
  protected void onFanOutReceive2(Message message, SMSBean sms, Channel channel) throws IOException {
    log.info("=========onFanOutReceive2已接收===========");
    if (consumer!=null) {
      consumer.accept("FanOut-"+SMSConstants.SMS_QUEUE_FANOUT_2, sms);
    }
  }
  @Override
  public void setOnMessage(BiConsumer<String,SMSBean> consumer) {
    this.consumer=consumer;
  }
}
